package com.sjc.lesson02.api.broadcast;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * broadcast广播变量
 * 需求：
 *      flink会从数据源中获取到用户的姓名
 *      最终需要把用户的姓名和年龄信息打印出来
 * 分析：
 *      所以就需要在中间的map处理的时候获取用户的年龄信息
 *      建议把用户的关系数据集使用广播变量进行处理
 */
public class BroadCastDemo {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 1：准备需要广播的数据
        ArrayList<Tuple2<String, Integer>> broadcast = new ArrayList<>();
        broadcast.add(new Tuple2<>("zs", 18));
        broadcast.add(new Tuple2<>("ls", 20));
        broadcast.add(new Tuple2<>("ww", 17));

        DataSource<Tuple2<String, Integer>> tupleData = env.fromCollection(broadcast);

        //1.1:处理需要广播的数据,把数据集转换成map类型，map中的key就是用户姓名，value就是 用户年龄
        MapOperator<Tuple2<String, Integer>, HashMap<String, Integer>> toBroadcast =
                tupleData.map(new MapFunction<Tuple2<String, Integer>, HashMap<String, Integer>>() {
            @Override
            public HashMap<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                HashMap<String, Integer> res = new HashMap<>();
                res.put(value.f0, value.f1);
                return res;
            }
        });

        // 源数据
        DataSource<String> data = env.fromElements("zs", "ls", "ww");

        // 注意：在这里需要使用到RichMapFunction获取广播变量
        MapOperator<String, String> result = data.map(new RichMapFunction<String, String>() {
            List<HashMap<String, Integer>> broadCastMap = new ArrayList<>();
            HashMap<String, Integer> allMap = new HashMap<>();

            /**
             * 这个方法只会执行一次
             * 可以在这里实现一些初始化的功能
             * 所以，就可以在open方法中获取广播变量数据
             */
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                this.broadCastMap = getRuntimeContext().getBroadcastVariable("broadCastMapName");
                for (HashMap map : broadCastMap) {
                    allMap.putAll(map);
                }
            }

            @Override
            public String map(String value) throws Exception {
                Integer age = allMap.get(value);
                return value + " : " + age;
            }
        }).withBroadcastSet(toBroadcast, "broadCastMapName"); //2：执行广播数据的操作

        result.print();
    }
}
